package com.atguigu.day08;

import com.atguigu.util.ClickEvent;
import com.atguigu.util.ClickSource;
import com.atguigu.util.ProductViewCountPerWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Example6 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ClickEvent>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                            @Override
                            public long extractTimestamp(ClickEvent clickEvent, long l) {
                                return clickEvent.timestamp;
                            }
                        })
                )
                .keyBy(r -> r.username)
                .window(TumblingEventTimeWindows.of(Time.days(1)))
                .trigger(new Trigger<ClickEvent, TimeWindow>() {
                    // 在窗口的第一条数据到来之后的每个整数秒触发一次窗口计算
                    // 第一条数据来了以后注册一个接下来的整数秒的定时器
                    @Override
                    public TriggerResult onElement(ClickEvent clickEvent, long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        // 标志位来判断来的数据是否是第一条数据
                        // 窗口状态变量
                        ValueState<Boolean> flag = triggerContext.getPartitionedState(
                                new ValueStateDescriptor<Boolean>(
                                        "flag",
                                        Types.BOOLEAN
                                )
                        );

                        if (flag.value() == null) {
                            // 来的是第一条数据
                            // 注册一个接下来的整数秒的定时器
                            // 1234ms接下来的整数秒怎么计算？
                            // 1234 + 1000 - 1234 % 1000
                            long nextSecond = clickEvent.timestamp + 1000 - clickEvent.timestamp % 1000;
                            triggerContext.registerEventTimeTimer(nextSecond); // 注册的是下面的onEventTime定时器
                            flag.update(true);
                        }

                        return TriggerResult.CONTINUE;
                    }

                    @Override
                    public TriggerResult onProcessingTime(long l, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        return null;
                    }

                    @Override
                    public TriggerResult onEventTime(long timerTs, TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        // timerTs参数是注册的定时器的时间戳
                        if (timerTs < timeWindow.getEnd() - 1) {
                            if (timerTs + 1000L < timeWindow.getEnd() - 1) {
                                // 注册下个整数秒的定时器
                                // 这里注册的还是onEventTime定时器
                                triggerContext.registerEventTimeTimer(timerTs + 1000L);
                            }
                            return TriggerResult.FIRE; // 触发窗口执行
                        // flink会默认注册一个窗口结束时间-1毫秒的onEventTime定时器
                        } else if (timerTs == timeWindow.getEnd() - 1) {
                            // 触发窗口计算并清空窗口
                            return TriggerResult.FIRE_AND_PURGE;
                        } else {
                            return TriggerResult.CONTINUE;
                        }
                    }

                    @Override
                    public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {
                        // 单例
                        ValueState<Boolean> flag = triggerContext.getPartitionedState(
                                new ValueStateDescriptor<Boolean>(
                                        "flag",
                                        Types.BOOLEAN
                                )
                        );

                        // 窗口销毁时清空状态
                        flag.clear();
                    }
                })
                .process(new ProcessWindowFunction<ClickEvent, ProductViewCountPerWindow, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<ClickEvent> iterable, Collector<ProductViewCountPerWindow> collector) throws Exception {
                        collector.collect(new ProductViewCountPerWindow(
                                key,
                                iterable.spliterator().getExactSizeIfKnown(),
                                context.window().getStart(),
                                context.window().getEnd()
                        ));
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
